package com.lemo.mapreduce;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Pattern;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.lemo.mapreduce.model.Gift;
import com.lemo.mapreduce.model.GiftSort;
import com.lemo.mapreduce.model.TableJoin;

/**
 * 双表联合查询
 * @author lemo
 *
 */
public class mapredJoin extends Configured implements Tool
{
    public static void main( String[] args ) throws Exception
    {
    	args = new String[]{ //
    			"/user/beifeng/tmp" ,//
    			"/user/beifeng/output"
    		};
    	Configuration conf = new Configuration();
    	int status = ToolRunner.run(conf, new mapredJoin(), args);
    	System.exit(status);
    }

	@Override
	public int run(String[] arg0) throws Exception {
		// TODO Auto-generated method stub
		Configuration conf = this.getConf();
		Job job = Job.getInstance(conf,this.getClass().getSimpleName());
		job.setJarByClass(this.getClass());
		
		Path inpath = new Path(arg0[0]);
		FileInputFormat.addInputPath(job, inpath);
		//map
		job.setMapperClass(MapperClass.class);
		job.setMapOutputKeyClass(IntWritable.class);
		job.setMapOutputValueClass(TableJoin.class);
		
		//reduce
		job.setNumReduceTasks(1);
		job.setReducerClass(ReducerClass.class);
		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(Text.class);
		
		
		Path outpath = new Path(arg0[1]);
		FileOutputFormat.setOutputPath(job, outpath);
		
		boolean sucess = job.waitForCompletion(true);
		return sucess? 0 : 1 ;
	}
    
	public static class MapperClass extends Mapper<LongWritable, Text, IntWritable, TableJoin> {

		private TableJoin mapOutputValue = new TableJoin();
		private IntWritable mapOutputKey = new IntWritable();
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			
			String lineValue = value.toString();
			
			// split
			String[] strs = lineValue.split(",") ;
			if(strs.length!=12&&strs.length!=5) {
				return;
			}else if(strs.length == 12) {
				Gift gift = new Gift(strs[0],strs[1],strs[2],strs[3],strs[4],strs[5],strs[6],strs[7],strs[8],strs[9],strs[10],strs[11]);
				mapOutputKey.set(gift.getGid());//将gid设置成key
				mapOutputValue.set("gift", gift.toString());
			}else if(strs.length == 5) {
				GiftSort gitfSort = new GiftSort(strs[0],strs[1],strs[2],strs[3],strs[4]);
				mapOutputKey.set(gitfSort.getId());
				mapOutputValue.set("gitfSort", gitfSort.toString());
			}
			context.write(mapOutputKey, mapOutputValue);
		}
		
	}
	public static class ReducerClass extends Reducer<IntWritable, TableJoin, NullWritable, Text>{

		private Text outputValue = new Text();
		@Override
		protected void reduce(IntWritable key, Iterable<TableJoin> values,
				Context context) throws IOException, InterruptedException {
			String sort = "";
			List<String> gifts = new ArrayList<String>();
			for(TableJoin value:values) {
				if("gift".equals(value.getTag())) {
					gifts.add(value.getInfo());
				}else if("gitfSort".equals(value.getTag())) {
					sort = value.getInfo();
				}
			}
			for(String gift:gifts) {
				outputValue.set(gift+sort);
				context.write(null,outputValue);
			}
		}
		
	}
    
}